モックが無くてLambdaのテストが辛い?!無ければ自分で作ればいい!! Greengrassのユニットテストが辛かったのでmotoを拡張してGreengrassのモックを実装してみた
はじめに
サーバーレス開発部＠大阪の岩田です。 最近Greengrassを活用したシステムのバックエンド開発をやっていのですが、ローカル環境でのユニットテストがとても辛いです。いや、辛かったです。 DynamoDBのようなメジャーサービスであればDynamoDB LocalやLocalStackを利用することで、比較的簡単にユニットテストを回していくことができますが、GreengrassにはDynamoDB Localのようなモックツールが存在せず、LocalStackもGreengrassには未対応となっています。
解決策としてmotoを拡張してGreengrassのモックを作成したので手順についてご紹介します。 以前紹介した下記ブログでは既にmoto本体に実装済みのDynamoDBのモックにTransactWriteItemsの機能を追加しましたが、今回はmoto本体にモックが実装されていないサービスへのモックの実装となります。
AWSのサービスをモックするライブラリmotoを拡張してDynamoDBのTransactWriteItemsを実装する
なおソースコードはGitHubで公開しています。
環境
以下の環境で開発中のプロジェクトにGreengrassのモックを追加しました。
- OS: Mac OS X 10.14.2
- Python: 3.6.5
- boto3: 1.9.71
- botocore: 1.12.71
- moto: 1.3.7
motoを拡張する手順
今回はtestsディレクトリにmotoというディレクトリを作成し、さらにその中のgreengrassというディレクトリに諸々の処理を実装しました。
/tests/ ├── moto │ └── greengrass │ ├── __init__.py │ ├── exceptions.py │ ├── models.py │ ├── responses.py │ └── urls.py
ざっくりと各ファイルの役割は下記の通りです。
__init__.pyモック用のモジュールを初期化するファイルexceptions.pyGreengrassの独自例外を定義するファイルmodels.pyモックのメインロジックを実装するファイルresponses.pyboto3から受け付けたAPIリクエストを受け付けてmodels.pyの適切な処理に振り分けるファイルurls.pyGreengrassのAPIリクエスト先のURLを定義するファイル
各ファイルの詳細を見ていきながら、boto3のGreengrass Clientにcreate_groupのモックを実装する手順を解説します。
init.py
from __future__ import unicode_literals from .models import greengrass_backends from moto.core.models import base_decorator greengrass_backend = greengrass_backends['us-east-1'] mock_greengrass = base_decorator(greengrass_backends)
このファイルでやることはシンプルです。mock_greengrassという名前でGreengrassバックエンドのモックを準備します。
これでgreengrassモジュールをimportしたファイルからはmock_greengrassという名前でモックにアクセスできるようになります。
urls.py
from __future__ import unicode_literals
from .responses import GreengrassResponse
url_bases = [
"https?://greengrass.(.+).amazonaws.com",
]
response = GreengrassResponse()
url_paths = {
'{0}/.*$': response.dispatch,
}
このファイルではGreengrassAPIのエンドポイントURLを定義します。エンドポイントの一覧はGreengrassのAPI Referenceに記載されています。が、Greengrassのエンドポイントはシンプルなので単にurl_basesにhttps?://greengrass.(.+).amazonaws.comを入れておけばOKです。
ここで定義したURLに対してboto3がリクエストを発行した際にmotoのBaseResponseで定義されたdispatchメソッドに処理が渡り、処理がモックに置き換わるという流れです。dispatchメソッドは後述するresponses.pyで定義した処理にリクエストをルーティングします。
exceptions.py
このファイルではboto3のClientErrorのエラーコード毎に独自例外を定義します。 boto3では例外をClientErrorというクラスで表現しています。例えばこんな感じです。
>>> client.create_group(Name='')
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/Users/xxxxxx/site-packages/botocore/client.py", line 357, in _api_call
return self._make_api_call(operation_name, kwargs)
File "/Users/xxxxxx/site-packages/botocore/client.py", line 661, in _make_api_call
raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidContainerDefinition) when calling the CreateGroup operation: InvalidContainerDefinitionException: Group name is missing in the input
このClientErrorクラスはresponseというプロパティを持っており、このプロパティの中でさらに詳細なエラーメッセージやエラーコードを表現しています。responseの中身はこんな感じで、先ほど発生させた例外であればエラーコードはInvalidContainerDefinitionとなります。
{'Error': {'Message': 'InvalidContainerDefinitionException: Group name is missing in the input', 'Code': 'InvalidContainerDefinition'}, 'ResponseMetadata': {'RequestId': '5fb1d2f6-56a6-11e9-b19b-bb2f7987dd15', 'HTTPStatusCode': 400, 'HTTPHeaders': {'date': 'Thu, 04 Apr 2019 06:53:38 GMT', 'content-type': 'application/json', 'content-length': '132', 'connection': 'keep-alive', 'x-amzn-requestid': '5fb1d2f6-56a6-11e9-b19b-bb2f7987dd15', 'x-amzn-greengrass-trace-id': 'Root=1-5ca5a9f2-4eb3d60f600f734096d4be3f', 'x-amz-apigw-id': 'Xmd91HzptjMFfHw=', 'x-amzn-trace-id': 'Root=1-5ca5a9f2-995be03e61145afc4c2569b2;Sampled=0'}, 'RetryAttempts': 0}}
エラーコード:InvalidContainerDefinitionExceptionを表現する場合はこんな感じです。
from __future__ import unicode_literals
from moto.core.exceptions import JsonRESTError
class GreengrassClientError(JsonRESTError):
code = 400
class InvalidContainerDefinitionException(GreengrassClientError):
def __init__(self, msg):
self.code = 400
super(InvalidContainerDefinitionException, self).__init__(
"InvalidContainerDefinitionException",
msg
)
まずmoto.core.exceptions.JsonRESTErrorを継承したGreengrassClientErrorを定義し、さらにGreengrassClientErrorを継承したInvalidContainerDefinitionExceptionを定義します。
例外クラスのコンストラクタではcodeにHTTPのステータスコードを設定します。大体400か404です。設定すべきステータスコードは実際にboto3で例外を発生させて確認して下さい。
responses.py
このファイルではboto3から発行されたリクエストを自作したモックのバックエンドに引き渡す中継処理を実装します。
from __future__ import unicode_literals
import json
from moto.core.responses import BaseResponse
from tests.moto.greengrass.models import greengrass_backends
class GreengrassResponse(BaseResponse):
SERVICE_NAME = "greengrass"
@property
def greengrass_backend(self):
return greengrass_backends[self.region]
def create_group(self):
name = self._get_param("Name")
initial_version = self._get_param("InitialVersion")
res = self.greengrass_backend.create_group(name=name, initial_version=initial_version)
return 201, {"status": 201}, json.dumps(res.to_dict())
まずBaseResponseクラスを継承した独自のResponseクラスを作成します。
作成したらSERVICE_NAMEという変数にAWSのサービス名を設定します。今回の例だとgreengrassです。
次にgreengrass_backendというプロパティを定義し、
ここまでできたら後はboto3クライアントの各メソッドに対応したメソッドを定義していきます。今回の例ではcreate_groupを使います。この処理の中では
BaseResponseクラスの_get_paramメソッドを利用してboto3の各メソッドに渡された名前付き引数の値を取得- モックのバックエンド処理実行
- レスポンスの返却
といった操作を行います。
models.pyの追加
このファイルにモックのメイン処理を実装していきます。
Fakexxxxxクラスの追加
対象AWSサービスが管理するエンティティを表現するためにFakexxxxxというクラスを自作します。
今回はcreate_groupを実装してGreengrassのGroupを作成するので、GreengrassのGroupを表現するエンティティを作成します。
class FakeGroup(BaseModel):
def __init__(self, region_name, name):
self.region_name = region_name
self.group_id = str(uuid.uuid4())
self.name = name
self.arn = "arn:aws:greengrass:%s:1:/greengrass/groups/%s" % (self.region_name, self.group_id)
self.created_at_datetime = datetime.utcnow()
self.last_updated_datetime = datetime.utcnow()
self.latest_version = ''
self.latest_version_arn = ''
def to_dict(self):
res = {
"Arn": self.arn,
"CreationTimestamp": self.created_at_datetime.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
"Id": self.group_id,
"LastUpdatedTimestamp": self.last_updated_datetime.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
"Name": self.name,
}
if self.latest_version:
res["LatestVersion"] = self.latest_version
res["LatestVersionArn"] = self.latest_version_arn
return res
class FakeGroupVersion(BaseModel):
def __init__(self, region_name, group_id, definition):
self.region_name = region_name
self.group_id = group_id
self.version = str(uuid.uuid4())
self.arn = "arn:aws:greengrass:%s:1:/greengrass/groups/%s/versions/%s" \
% (self.region_name, self.group_id, self.version)
self.created_at_datetime = datetime.utcnow()
self.definition = definition
def to_dict(self):
return {
"Arn": self.arn,
"CreationTimestamp": self.created_at_datetime.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
"Definition": self.definition,
"Id": self.group_id,
"Version": self.version
}
Fakexxxxxクラスには
- コンストラクタ
- APIのモックレスポンス返却用に各メンバ変数を辞書形式で返却する処理
を実装します。
GreengrassBackendクラスの実装
ここまでできたら、モックのメインを作成します。
moto.core.BaseBackendを継承してGreengrassBackendというクラスを作成し、諸々の処理を実装していきます。
class GreengrassBackend(BaseBackend):
def __init__(self, region_name):
super(GreengrassBackend, self).__init__()
self.region_name = region_name
self.groups = OrderedDict()
self.group_versions = OrderedDict()
def reset(self):
region_name = self.region_name
self.__dict__ = {}
self.__init__(region_name)
def create_group(self, name, initial_version):
if name == "":
raise InvalidContainerDefinitionException(
"Group name is missing in the input"
)
group = FakeGroup(self.region_name, name)
self.groups[group.group_id] = group
if initial_version:
self.create_group_version(group.group_id, initial_version)
return group
def create_group_version(self, group_id, definition):
# TODO Implement validation
group_ver = FakeGroupVersion(self.region_name, group_id, definition)
group_vers = self.group_versions.get(group_ver.group_id, {})
group_vers[group_ver.version] = group_ver
self.group_versions[group_ver.group_id] = group_vers
self.groups[group_id].latest_version_arn = group_ver.arn
self.groups[group_id].latest_version = group_ver.version
return group_ver
まずはコンストラクタです。
モックのバックエンドクラスではAWSサービスの各エンティティをOrderedDictで管理します。コンストラクトの中にメンバ変数をOrderedDictで初期化するロジックを入れていきましょう。
コンストラクタが実装できたら後は実際のバックエンド処理を実装します。 ココが一番の腕の見せ所です。boto3の挙動をよーく観察し、Greengrassのサービスが裏で何をしているのか思いを馳せながら適切な処理を実装します。
最後に他のファイルからimportしてもらうためにgreengrass_backendsを定義します。
available_regions = boto3.session.Session().get_available_regions("greengrass")
greengrass_backends = {region: GreengrassBackend(region) for region in available_regions}
これでモックの作成は終了です！
自作したモックを使ってみる
モックが自作できたの使ってみましょう！ 今回のPJではpytestのfixture内でAWSサービスのモック開始・停止を行なっています。 greengrassのモック処理だけ自作したファイルからimportしてこんな感じで使います。
import boto3
import moto
import pytest
from unittest import mock
from tests.moto.greengrass import mock_greengrass
@pytest.fixture()
def start_moto_mock():
moto.mock_iot().start()
moto.mock_sts().start()
moto.mock_iotdata().start()
moto.mock_dynamodb2().start()
moto.mock_s3().start()
mock_greengrass().start()
moto.mock_lambda().start()
yield
moto.mock_iot().stop()
moto.mock_sts().stop()
moto.mock_iotdata().stop()
moto.mock_dynamodb2().stop()
moto.mock_s3().stop()
mock_greengrass().stop()
moto.mock_lambda().stop()
#...略
テスト対象となるLambda関数のコードです。boto3を利用して入力値を元にGreengrassグループを作成するだけの簡単な処理です。
import boto3
import json
gg_client = boto3.client("greengrass", region_name="ap-northeast-1")
def handler(event, context):
grp_name = event['body']['grp_name']
res = gg_client.create_group(Name=grp_name)
return json.dumps({
"statusCode": 201,
"body": res
})
テストコード本体です
from botocore.exceptions import ClientError
import json
import pytest
from src.create_gg_group import handler
@pytest.mark.usefixtures('start_moto_mock')
class TestClass(object):
def test_create_gg_group(self):
event = {
"body": {
"grp_name": "test_grp"
}
}
res = handler(event, {})
data = json.loads(res)
assert data["statusCode"] == 201
assert data["body"]["Name"] == "test_grp"
def test_empty_group_name(self):
event = {
"body": {
"grp_name": ""
}
}
with pytest.raises(ClientError) as ex:
handler(event, {})
assert ex.value.response["Error"]["Message"] == "Group name is missing in the input"
assert ex.value.response["Error"]["Code"] == "InvalidContainerDefinitionException"
実際にpytestでGreengrass周りのテストを実行したログです
platform darwin -- Python 3.6.5, pytest-4.4.1, py-1.8.0, pluggy-0.9.0 -- /Users/xxxxxxx/.venv/bin/python cachedir: .pytest_cache rootdir: /Users/xxxxxxx/ collected 2 items tests/unit/test_create_gg_group.py::TestClass::test_create_gg_group PASSED [ 50%] tests/unit/test_create_gg_group.py::TestClass::test_empty_group_name PASSED
バッチリ実行できています。
まとめ
自力でモック作るのはかなりパワーがかかりましたが、その甲斐あってユニットテストはかなり楽になりました。きっとこの頑張りがPJの中盤ぐらいからボディブローのように効いてくるはず。 また、モックを実装するに当たってGreengrassというサービスが内部的にどうのようにデータを保持しているのか、各エンティティの関連はどうなっているのか？ということを分析したので、Greengrassというサービスにもそこそこ詳しくなれたかなと感じています。これもモックを自力実装して良かった点です。 自作したモックはまだ品質的に不安な面もあるので、もう少しプロジェクトを進めながらブラッシュアップした後でmoto本体にプルリクを上げたいと思います。
今回紹介した手順の大枠は他のAWSサービスにも流用できるはずなので、AWSサービスを使用するアプリのユニットテストでお困りの方がいれば、motoの自力拡張を検討してみてはいかがでしょうか？
![[アップデート] AWS Lambda で Python 3.13 ランタイムがサポートされました](https://images.ctfassets.net/ct0aopd36mqt/wp-thumbnail-484d8afca1cbe50513cd4d6738af2960/fa0838fa40f247093d532945fc785e88/aws-lambda)
